package com.zallds.watermark;

import lombok.Data;
import lombok.NoArgsConstructor;
import org.apache.commons.lang3.StringUtils;
import org.apache.commons.lang3.time.FastDateFormat;
import org.apache.flink.api.common.eventtime.*;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.WindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;

import java.io.Serializable;
import java.time.Duration;
import java.util.ArrayList;
import java.util.List;

/**
 * 对于延迟的数据,使用侧输出流
 */
public class WaterMarkSideOutputLateDataDemo {
    static FastDateFormat df = FastDateFormat.getInstance("HH:mm:ss");


    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.createLocalEnvironmentWithWebUI(new Configuration());
        env.setParallelism(1);


        OutputTag<Order> seriousLateOutputTag = new OutputTag<Order>("seriousLate", TypeInformation.of(Order.class));

        SingleOutputStreamOperator<Order> sourceDS = env.socketTextStream("192.168.62.37", 8889)
                .filter(keyword -> StringUtils.isNotEmpty(keyword))
                .map(new Splitter());


        SingleOutputStreamOperator<String> result = sourceDS.assignTimestampsAndWatermarks(
                WatermarkStrategy.<Order>forBoundedOutOfOrderness(Duration.ofSeconds(3))
                        .withTimestampAssigner((order, timestamp) -> order.getEventTime())
        )
                .keyBy(Order::getName)
                .window(TumblingEventTimeWindows.of(Time.seconds(5)))
                .sideOutputLateData(seriousLateOutputTag)
                .apply(new MyWindowFunction());
        result.print("正常输出->");
        DataStream<Order> sideOutput = result.getSideOutput(seriousLateOutputTag);
        sideOutput.print("延迟输出->");

        env.execute();

    }

    private static class Splitter implements MapFunction<String, Order> {
        @Override
        public Order map(String value) throws Exception {
            String name = value.split(" ")[0];
            Integer money = Integer.parseInt(value.split(" ")[1]);
            Long eventTime = Long.parseLong(value.split(" ")[2]);
            return new Order(name, money, eventTime);
        }
    }


    @Data
    @NoArgsConstructor
    public static class Order implements Serializable {
        private String name;
        private Integer money;
        private Long eventTime;

        public Order(String name, Integer money, Long eventTime) {
            System.out.println(String.format("原始数据 %s , %s , %s ", name, money, df.format(eventTime)));
            this.name = name;
            this.money = money;
            this.eventTime = eventTime;
        }
    }


    private static class MyWindowFunction implements WindowFunction<Order, String, String, TimeWindow> {

        @Override
        public void apply(String key, TimeWindow window, Iterable<Order> orders, Collector<String> out) throws Exception {
            List<String> list = new ArrayList<>();
            for (Order order : orders) {
                Long eventTime = order.eventTime;
                String formatEventTime = df.format(eventTime);
                list.add(formatEventTime);
            }
            String start = df.format(window.getStart());
            String end = df.format(window.getEnd());
            //现在就已经获取到了当前窗口的开始和结束时间,以及属于该窗口的所有数据的事件时间,把这些拼接并返回
            String outStr = String.format("key:%s,窗口开始结束:[%s~%s),属于该窗口的事件时间:%s", key, start, end, list.toString());
            out.collect(outStr);
        }
    }
}
